package com.shujia.flink.transformation

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo1Map {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds1: DataStream[String] = env.socketTextStream("master", 9999)

    //
    //    ds1.map(new MapFunction[String, String] {
    //      override def map(value: String): String = {
    //        value
    //      }
    //    })
    //      .print()


    ds1
      .flatMap(new TestFlatMapFunction)
      .print()


    env.execute()

  }

}

class TestFlatMapFunction extends RichFlatMapFunction[String, String] {


  //open方法在每一个分区中只会执行异常
  override def open(parameters: Configuration): Unit = {

    //在这里创建mysql连接 每一个task只会创建异常

    println("open")
  }

  override def close(): Unit = {

  }

  override def flatMap(value: String, out: Collector[String]): Unit = {

    //如果在这里mysql连接，每一条数据都会创建一个连接


    value
      .split(",")
      .foreach(word => {

        //发送到下游
        out.collect(word)

      })

  }
}
